package de.preissler.selection;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.RangeSlicesQuery;

public class Main
{
	private static final String DATE_SOURCE = "DAX";
	private static final String KEYSPACE_NAME = "Exchange";

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception
	{
		//Cluster initialisieren
		Cluster cluster = getCluster();

		Keyspace keyspaceOperator = HFactory.createKeyspace(KEYSPACE_NAME, cluster);
		keyspaceOperator.setConsistencyLevelPolicy(new me.prettyprint.cassandra.model.AllOneConsistencyLevelPolicy());
		
		//Tage holen fuer die die Gewinner ermittelt werden sollen
		List<String> dates = getDates(keyspaceOperator);
		
		//Anzahl der Gewinner erfagen
		int numberOfWinners = getNumberOfWinners();
		
		//Konkrete Instanz laden
		Winners winners = getInstanceOfWinners();
		
		//speichert die gemessene Zeit
		long[] time = new long[3];
		
		//speichert das Ergebnis
		Map<String, List<ChangeOfISIN>> result = new HashMap<String, List<ChangeOfISIN>>(dates.size());
		
		//das ganze 3 mal machen
		for(short i = 0; i < 3; i++)
		{
			long start = System.nanoTime();
			
			result = winners.getWinners(dates, keyspaceOperator, result, numberOfWinners);
			
			long end = System.nanoTime();
			
			time[i] = end - start;
		}
		
		//Hector sauber beenden
		cluster.getConnectionManager().shutdown();

		for(long i : time)
		{
			System.out.println(i + " ns");
		}
		System.out.println("Size: " + result.size());
		System.out.println(result);
	}
	
	private static Winners getInstanceOfWinners() throws Exception
	{
		System.out.println("<1> View   or   <2> Calculate    ?");
		int method = Integer.valueOf(readLine());
		if(method == 1)
			return new WinnersView();
		else
			return new WinnersCalc();
	}
	
	/*
	 * Holt alle Keys raus
	 */
	public static List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}
	
	private static List<String> getDates(Keyspace keyspace) throws Exception
	{
		System.out.println("How many dates should be calculated?");
		int n = Integer.valueOf(readLine());
		List<String> dates = getKeys(keyspace, DATE_SOURCE);
		List<String> result = new LinkedList<String>();
		
		for(int i = dates.size()-1; i >= dates.size()-n; i--)
		{
			result.add(dates.get(i));
		}
		
		return result;
	}
	
	private static int getNumberOfWinners() throws Exception
	{
		System.out.println("How many winners per date would you like to have?");
		return Integer.valueOf(readLine());
	}
	
	private static String readLine() throws Exception
	{
		BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
		return in.readLine();
	}
	
	/*
	 * Erstellt das Cluster Objekt
	 */
	private static Cluster getCluster() throws Exception
	{
		String clusterName;
		String host;
		String port;
		
		System.out.println("In order to connect to cassandra we need some parameters...");
		
		System.out.print("clustername: ");
		clusterName = readLine();
		
		System.out.print("host: ");
		host = readLine();
		
		System.out.print("port: ");
		port = readLine();
		
		return HFactory.getOrCreateCluster(clusterName, host + ":" + port);
	}

}
